package cn.hy.dataSync.transport.exchanger;


import cn.hy.dataSync.config.*;
import cn.hy.dataSync.element.Column;
import cn.hy.dataSync.element.Record;
import cn.hy.dataSync.statistics.communication.Communication;
import cn.hy.dataSync.statistics.communication.CommunicationTool;
import cn.hy.dataSync.transport.channel.Channel;
import cn.hy.dataSync.transport.record.TerminateRecord;
import cn.hy.dataSync.utils.ExpressUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.Validate;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.Optional;

/**
 * 读写数据交换
 */
@Slf4j
public class RecordExchanger implements RecordSender, RecordReceiver {

    private final RuntimeException shutdownException =
            new RuntimeException("Task 收到了shutdown 指令，为 failover 做准备");

    private final AbstractTaskDirtyCollector dirtyCollector;

    private final Communication currentCommunication;

    private final Channel channel;

    private final List<Transformer> transformers;

    private static Class<? extends Record> RECORD_CLASS;

    private final List<Record> buffer;

    private int bufferSize;

    private int bufferIndex = 0;

    private volatile boolean shutdown = false;

    private long totalExhaustedTime = 0;
    private long totalFilterRecords = 0;
    private long totalSuccessRecords = 0;
    private long totalFailedRecords = 0;

    public RecordExchanger(final Channel channel,
                           final Communication communication,
                           final AbstractTaskDirtyCollector dirtyCollector) {
        this(channel, communication, null, dirtyCollector);
    }

    @SuppressWarnings("unchecked")
    public RecordExchanger(final Channel channel,
                           final Communication communication,
                           final List<Transformer> transformers,
                           final AbstractTaskDirtyCollector dirtyCollector) {
        assert null != channel;
        this.channel = channel;
        this.transformers = Optional.ofNullable(transformers)
                .orElse(new ArrayList<>());
        this.bufferSize = channel.getCoreConfig()
                .getInteger(CoreConstant.EXCHANGER_BUFFER_SIZE, 10);
        this.currentCommunication = communication;
        this.dirtyCollector = dirtyCollector;
        this.buffer = new ArrayList<>(bufferSize);
        try {
            RECORD_CLASS = (Class<? extends Record>) Class
                    .forName(channel.getCoreConfig().getString(
                            CoreConstant.TASK_CORE_TRANSPORT_RECORD_CLASS,
                            CoreConstant.TASK_CORE_TRANSPORT_RECORD_DEFAULT));
        } catch (ClassNotFoundException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public Record createRecord() {
        try {
            return RECORD_CLASS.newInstance();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void sendToWriter(Record record) {
        if (shutdown) {
            throw shutdownException;
        }
        Validate.notNull(record, "record不能为空");

        // record 数据 记录转换
        record = doTransformer(record);

        if (Objects.isNull(record)) {
            return;
        }

        boolean isFull = (this.bufferIndex >= this.bufferSize);
        if (isFull) {
            flush();
        }

        this.buffer.add(record);
        this.bufferIndex++;
    }

    @Override
    public void flush() {
        if (shutdown) {
            throw shutdownException;
        }
        // 将数据写入 channel（通道） 中
        this.channel.pushAll(this.buffer);
        this.doStat(); // 记录信息
        this.buffer.clear();
        this.bufferIndex = 0;
    }

    @Override
    public void terminate() {
        if (shutdown) {
            throw shutdownException;
        }
        flush();
        this.channel.pushTerminate(TerminateRecord.get());
    }

    @Override
    public void shutdown() {
        shutdown = true;
        try {
            buffer.clear();
            channel.clear();
        } catch (Throwable t) {
            t.printStackTrace();
        }
    }

    @Override
    public Record getFromReader() {
        if (shutdown) {
            throw shutdownException;
        }
        boolean isEmpty = (this.bufferIndex >= this.buffer.size());
        if (isEmpty) {
            receive();
        }

        Record record = this.buffer.get(this.bufferIndex++);
        if (record instanceof TerminateRecord) {
            record = null;
        }
        return record;
    }

    private void receive() {
        this.channel.pullAll(this.buffer);
        this.bufferIndex = 0;
        this.bufferSize = this.buffer.size();
    }

    /**
     * 数据转换
     * @param record record
     * @return Record
     */
    private Record doTransformer(Record record) {
        long diffExhaustedTime = 0;
        String errorMsg = null;
        boolean failed = false;

        for (Transformer transformer : transformers) {
            long startTs = System.nanoTime();

            try {
                Object value = null;
                // 判断是不是直接映射.
                if (transformer.isDirectMapping()) {
                    Column column = record.getColumn(transformer.getExpression());
                    if (column != null) {
                        value = column.getRawData();
                    }
                } else {
                    value = ExpressUtils.evaluate(transformer.getExpression(), record);
                }
                // 存储到内存中
                record.put(transformer.getColumn(), value);
            } catch (Exception e) {
                errorMsg = String.format("transformer has Exception(%s)", e.getMessage());
                failed = true;
                break;
            }

            long diff = System.nanoTime() - startTs;
            diffExhaustedTime += diff;
        }

        totalExhaustedTime += diffExhaustedTime;

        if (failed) {
            totalFailedRecords++;
            dirtyCollector.collectDirtyRecord(record, errorMsg);
            return null;
        } else {
            totalSuccessRecords++;
            return record;
        }
    }

    private void doStat() {
        currentCommunication.setLongCounter(CommunicationTool.TRANSFORMER_SUCCEED_RECORDS, totalSuccessRecords);
        currentCommunication.setLongCounter(CommunicationTool.TRANSFORMER_FAILED_RECORDS, totalFailedRecords);
        currentCommunication.setLongCounter(CommunicationTool.TRANSFORMER_FILTER_RECORDS, totalFilterRecords);
        currentCommunication.setLongCounter(CommunicationTool.TRANSFORMER_USED_TIME, totalExhaustedTime);
    }
}
